package com.mitake.task.weekK;

import java.util.Arrays;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.mitake.network.HttpData;
import com.mitake.network.ISocketListener;
import com.mitake.network.MitakeHttpParams;
import com.mitake.network.MitakeSocket;
import com.mitake.parse.Day5;
import com.mitake.parse.WeekK;
import com.mitake.parse.MitakeDay5;
import com.mitake.parse.MitakeWeekK;
import com.mitake.task.TelegramGroup;
import com.mitake.utils.MitakeLog;
import com.mitake.utils.Util;

public class WeekKCompareTask {
	
	private static final String COLUMN_SEPARATOR = String.valueOf((char)0x02);
	private static int TIMEOUT = 1000 * 1000000;
	private static WeekKCompareTask instance = null;
	
	/**
	 * 封包序号
	 */
	private AtomicInteger packageNo;
	private int hostCount;
	private String[] stock;
	private String timestamp;
	private MitakeSocket[] sockets;
	private Hashtable<String, TelegramGroup> queue;
	private ConcurrentLinkedQueue<TelegramGroup> saveTelegramQueue;
	private long runInterval;
	private boolean running;
	private String logPath;
	
	/**
	 * 发送比数
	 */
	private long writeTelegramCount;
	
	/**
	 * 回传笔数
	 */
	private long readTelegramCount;
	
	private WeekKCompareTask(){};
	
	public static synchronized WeekKCompareTask getInstance(){
		if(instance == null){
			instance = new WeekKCompareTask();
		}
		return instance;
	}
	
	public void setRunInterval(long runInterval)
	{
		this.runInterval = runInterval;
	}
	
	public void setStockList(String[] stock)
	{
		this.stock = stock;
	}
	
	public void setLogPath(String logPath)
	{
		this.logPath = logPath;
	}

	
	public boolean isRunning()
	{
		return running;
	}
	
	public String[] getTelegramCount()
	{
		return new String[]{String.valueOf(writeTelegramCount), String.valueOf(readTelegramCount)};
	}
	
	public String getTimestamp() {
		return timestamp;
	}

	public void setTimestamp(String timestamp) {
		this.timestamp = timestamp;
	}

	public String[][] getAllSocketStatus()
	{
		String[][] data = null;
		if(sockets != null){
			data =new String[hostCount][2];
			for(int i=0;i<sockets.length; i++){
				data[i][0] = sockets[i].desc +"_"+sockets[i].ip +":"+sockets[i].port;
				data[i][1] = sockets[i].isConnected()?"0":"1";
				if(Util.DEBUG){
					MitakeLog.log("连线状态==" + sockets[i].ip + ":" + sockets[i].port + "_" + sockets[i].isConnected());
				}
			}
		}

		return data;
	}
	
	public class SendAPI implements Runnable{
		@Override
		public void run() {
			long beforeTime = 0;
			StringBuilder stocks =new StringBuilder();
			for(int i =0;i< stock.length;i++){
				stocks.append(stock[i]).append(",");
			}
			stocks.setLength(stocks.length()-1);
		    String list = stocks.toString();
		    String pn = null;
		    while(running){
		    	try{
			    	if(System.currentTimeMillis() - beforeTime > runInterval){
			    		pn = String.valueOf(packageNo.incrementAndGet());
			    		for(int i=0; i< hostCount ;i++){
			    			//组成报文
			    			MitakeHttpParams params =new MitakeHttpParams();
			    			params.api = "/v2/weekk";
			    			params.C2SDataType = MitakeHttpParams.C_S_DATA_TYPE_STRING;
			    			params.S2CDataType = MitakeHttpParams.S_C_DATA_TYPE_BYTES;
			    			params.headers = new String[][]{{"Token","MitakeWeb"},{"Symbol",list},{"timestamp","20161115"}};
			    			params.packageNo = pn;
			    			params.method = "GET";
			    			if(sockets[i].write(params)){
			    				synchronized(WeekKCompareTask.this){
			    					writeTelegramCount++;
			    				}
			    			}

			    		}
			    		
			    		beforeTime = System.currentTimeMillis();
			    	}
		    	}catch(Exception e){
		    		e.printStackTrace();
		    	}

		    	
		    	try {
					TimeUnit.SECONDS.sleep(10);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
		    }
		}
	}
	
	
	public class CompareTask implements Runnable{

		@Override
		public void run() {
			while(running){
				if(queue!= null && !queue.isEmpty()){
					for(Enumeration<TelegramGroup> e = queue.elements();e.hasMoreElements();){
						TelegramGroup telegramGroup =  e.nextElement();
						if(telegramGroup.count >= hostCount){
						    
							int len = telegramGroup.count;
							for(int i=0; i<len-1;i++){
								if(!Arrays.equals(telegramGroup.httpData[i].b, telegramGroup.httpData[i+1].b)){
									telegramGroup.hasDiff = true;
									break;
								}
							}
							
							//放入欲存储的queue中
							saveTelegramQueue.offer(telegramGroup);
							
							queue.remove(telegramGroup.packageNo);
						}else if(System.currentTimeMillis() - telegramGroup.initTime > TIMEOUT){
							//超过规定的时间没有返回就当有问题
							
							telegramGroup.hasDiff = true;
							saveTelegramQueue.offer(telegramGroup);
							queue.remove(telegramGroup.packageNo);
						}
					}
				}
				try
				{
					TimeUnit.MILLISECONDS.sleep(100);
				}
				catch(InterruptedException e)
				{
					e.printStackTrace();
				}
			}
		}
		
	}
	
	public class SaveTask implements Runnable{

		@Override
		public void run() {
			StringBuilder data =new StringBuilder();
			while(running){
				TelegramGroup telegramGroup = saveTelegramQueue.poll();
				if(telegramGroup == null){
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}else if(telegramGroup.count > 0){
					data.setLength(0);
					WeekK [] weekKs = new WeekK[telegramGroup.count];
					for(int i = 0 ; i < telegramGroup.count ; i++){
							try {
								String str = new String(telegramGroup.httpData[i].b,"UTF-8");
								WeekK weekk = MitakeWeekK.parseWeekK(str);
								weekKs[i] = weekk;
								weekKs[i].host = telegramGroup.httpData[i].host;
								weekKs[i].name = telegramGroup.httpData[i].name;
								weekKs[i].writeTime = String.valueOf(telegramGroup.httpData[i].sendTime);
								weekKs[i].readTime = String.valueOf(telegramGroup.httpData[i].responseTime);
								weekKs[i].responseTime = String.valueOf((telegramGroup.httpData[i].responseTime - telegramGroup.httpData[i].sendTime) / 1000000) + "ms";
								
							/*
							    week5s[i] = MitakeDay5.parseDay5(str);
							    week5s[i].host = telegramGroup.httpData[i].host;
							    week5s[i].name = telegramGroup.httpData[i].name;
							    week5s[i].writeTime = String.valueOf(telegramGroup.httpData[i].sendTime);
							    week5s[i].readTime = String.valueOf(telegramGroup.httpData[i].responseTime);
							    week5s[i].responseTime = String.valueOf((telegramGroup.httpData[i].responseTime - telegramGroup.httpData[i].sendTime) / 1000000) + "ms";
							    */
							} catch (Exception e) {
								e.printStackTrace();
							}							
					}
					
					try{
					  /*
						if(telegramGroup.hasDiff){
							
							Day5 [] list = new Day5[telegramGroup.count+1];
							list[0] =new Day5();
							System.arraycopy(week5s, 0, list, 1, week5s.length);
							//Day5CompareProcess.compare(list);
						}*/
						//if(telegramGroup.hasDiff){
							WeekK [] list = new WeekK[telegramGroup.count+1];
							list[0] =new WeekK();
							System.arraycopy(weekKs, 0, list, 1, weekKs.length);
							WeekKCompareProcess.compare(list);
						//}
					}catch(Exception e){
						e.printStackTrace();
					}
					
				}
				
				try
				{
					TimeUnit.MILLISECONDS.sleep(100);
				}
				catch(InterruptedException e)
				{
					e.printStackTrace();
				}
				
			}
		}
		
	}
	
	
	
	
	public void start(String[][] host){
		hostCount = host.length;
		/**
		 * 初始化封包序号
		 */
		packageNo = new AtomicInteger();
		sockets =new MitakeSocket[hostCount];
		queue =new Hashtable<String,TelegramGroup>();
		saveTelegramQueue = new ConcurrentLinkedQueue<TelegramGroup>();
		//建立所有机房的链接
		for(int i=0;i<host.length;i++){
			sockets[i] =new MitakeSocket(host[i], new ISocketListener() {
				
				@Override
				public void onNetworkStatusChanged(MitakeSocket mitakeSocket, String packageNo, String code) {
					if(code.equals("00")){
						//连线成功
						if(Util.DEBUG){
							MitakeLog.log("连线成功==" + mitakeSocket.ip + ":" + mitakeSocket.port + "==" + packageNo);
						}
					}else if(code.equals("01")){
						if(Util.DEBUG){
							MitakeLog.log("断线,无需自动连线==" + mitakeSocket.ip + ":" + mitakeSocket.port);
						}
					}else if(code.equals("02")){
                    	if(Util.DEBUG)
                    	{
                    		MitakeLog.log("无预期的断线==" + mitakeSocket.ip + ":" + mitakeSocket.port);
                    	}
                    	try
						{
							TimeUnit.MILLISECONDS.sleep(100);
						}
						catch(InterruptedException e)
						{
							e.printStackTrace();
						}
                    	
                    	mitakeSocket.start();
					}
					
				}
				
				@Override
				public void onContent(HttpData httpData) {
					// TODO Auto-generated method stub
					readTelegramCount++;
					HashMap<String, String> header =httpData.headers;
					String packageNo = header.get("k");
					/**
					 * 找不到封包就丢掉
					 */
					if(packageNo != null){
						if(httpData.responseTime - httpData.sendTime<= TIMEOUT){
							TelegramGroup telegramGroup = queue.get(packageNo);
							if(telegramGroup == null){
								telegramGroup = new TelegramGroup(hostCount);
								telegramGroup.count = 0;
								telegramGroup.packageNo = packageNo;
								telegramGroup.initTime = System.nanoTime();
								telegramGroup.stock = stock;
								queue.put(packageNo, telegramGroup);
							}
							telegramGroup.httpData[telegramGroup.count] = httpData;
							telegramGroup.count++;
						}
					}
				}
			});
			sockets[i].start();
		}
		
		running = true;
		Thread sendApiThread =new Thread(new SendAPI());
		sendApiThread.start();
		
		Thread compareTaskThread =new Thread(new CompareTask());
		compareTaskThread.start();
		
		Thread saveTaskThread =new Thread(new SaveTask());
		saveTaskThread.start();
	}
	
	public void stop()
	{
		running = false;
		
		if(sockets != null)
		{
			for(int i = 0 ; i < sockets.length ; i++)
			{
				if(sockets[i] != null)
				{
					sockets[i].disconnect("", false);
				}
			}
		}

		packageNo = null;
		
		queue.clear();
		queue = null;
	}
	
	
	
	public static void main(String[] args) {
		/**
		 * [{"type":"沪深L1","ip":"http://123.125.108.117","port":"22016","desc":"土城_轻量"},
		 * {"type":"沪深L1","ip":"http://180.163.112.216","port":"22016","desc":"外高桥_重量"},
		 * {"type":"沪深L1","ip":"http://58.63.252.23","port":"22016","desc":"广州_重量"},
		 * {"type":"沪深L1","ip":"http://140.207.241.151","port":"22016","desc":"金桥_轻量"}]
		 */
		WeekKCompareTask task =WeekKCompareTask.getInstance();
		task.setTimestamp("20161123");
		task.setStockList(new String[]{"600000.sh"});
		String [][] host =new String[][]{
			{"沪深L1","http://123.125.108.117","22016","土城_轻量"},
			{"沪深L1","http://180.163.112.216","22016","外高桥_重量"},
			{"沪深L1","http://58.63.252.23","22016","广州_重量"},
			{"沪深L1","http://140.207.241.151","22016","金桥_轻量"}
			};
			
		task.start(host);
		//WeekKCompareTask.SendAPI sendApi =task.new SendAPI();
		//new Thread(sendApi).start();
	}
	
	
	
	
	
}
